package com.ruoyi.datastatistics.service.impl;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.ruoyi.basedata.domain.RedisKeyName;
import com.ruoyi.basedata.domain.TerminalsDb;
import com.ruoyi.basedata.domain.vo.TerminalsDbProtocol;
import com.ruoyi.basedata.service.ITerminalsDbService;
import com.ruoyi.common.core.redis.RedisCache;
import com.ruoyi.common.utils.ExceptionUtil;
import com.ruoyi.common.utils.ParseDataUntils;
import com.ruoyi.datastatistics.domain.Realtimegatherdata;
import com.ruoyi.datastatistics.domain.TerminalHistoryData;
import com.ruoyi.datastatistics.service.IRealtimegatherdataService;
import com.ruoyi.datastatistics.service.ITerminalHistoryDataService;
import com.ruoyi.datastatistics.service.MqttHaiAnPaiShuiDbService;
import com.ruoyi.datastatistics.vo.Mqtt_HaiAnPaiShuiDb;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.math.BigDecimal;
import java.util.*;

@Service
@Slf4j
public class MqttHaiAnPaiShuiDbImpl implements MqttHaiAnPaiShuiDbService {
    @Autowired
    private ITerminalsDbService terminalsDbService;
    @Autowired
    private RedisCache redisCache;
    @Autowired
    private ITerminalHistoryDataService terminalHistoryDataService;
    @Autowired
    private IRealtimegatherdataService realtimegatherdataService;

    private static String factoryCode = "1002";

    @Override
    public void parseHaiAnPaiShuiDbFromMqtt(String msg){
        try{
            List<Mqtt_HaiAnPaiShuiDb> yc = JSON.parseArray(msg, Mqtt_HaiAnPaiShuiDb.class);
            System.out.println(msg);
            yc.sort(Comparator.comparing(Mqtt_HaiAnPaiShuiDb::getTerminalTime).reversed());//降序


            List<TerminalHistoryData> list = new ArrayList<TerminalHistoryData>();
            if(!yc.isEmpty()){
    //            TerminalsDbProtocol terminalsDb = redisCache.getCacheMapValue(RedisKeyName.terminalDbInfo+factoryCode, yc.get(0).getSn());
                TerminalsDbProtocol terminalsDb = new TerminalsDbProtocol();
                String addressKey = yc.get(0).getSn();
//                TerminalsDbProtocol terminalsDbTest = redisCache.getCacheMapValue(RedisKeyName.terminalDbInfo+ factoryCode, addressKey);
                if(!redisCache.hmhave(RedisKeyName.terminalDbInfo+factoryCode, addressKey)){
                    TerminalsDb tb = new TerminalsDb();
                    tb.setImei(yc.get(0).getImei());
                    tb.setCommunicateAddress(yc.get(0).getSn());
                    terminalsDb = terminalsDbService.selectTerminalsDbProtocolList(tb).get(0);
                    if (terminalsDb != null)
                    {
                        redisCache.setCacheMapValue(RedisKeyName.terminalDbInfo+ factoryCode, addressKey, terminalsDb);
                    }else{
                        log.error("设备不存在,IMEI：{}",tb.getImei());
                        return;
                    }

                }else {
                    terminalsDb = redisCache.getCacheMapValue(RedisKeyName.terminalDbInfo+ factoryCode, addressKey);
                }

                //判断阀门状态
                Integer valveOn = yc.get(0).getValveAllOn();
                Integer valveOff = yc.get(0).getValveAllOff();
                UpdateWrapper<TerminalsDb> terminalUpdateWrapper = Wrappers.update();
                terminalUpdateWrapper.eq("communicate_address", terminalsDb.getCommunicateAddress());
                terminalUpdateWrapper.eq("supplier_id", terminalsDb.getSupplierId());
                Integer valveStatus = terminalsDb.getValveStatus();
                if(valveOn==1 && valveOff==0){
                    valveStatus = 0;//开
                }else if(valveOn==0 && valveOff==1){
                    valveStatus = 1;//关
                }else{
                    log.info("｛｝阀门动作中",yc.get(0).getSn());
                }
                terminalUpdateWrapper.set("valve_status", valveStatus);
                boolean updateFlag =  terminalsDbService.update(terminalUpdateWrapper);//更新数据库阀门状态
                if(updateFlag){
                    terminalsDb.setValveStatus(valveStatus);
                    redisCache.setCacheMapValue(RedisKeyName.terminalDbInfo+ factoryCode, yc.get(0).getSn(),terminalsDb);//更新redis
                }

                for (Mqtt_HaiAnPaiShuiDb item:yc) {
                    boolean flag = checkIsSaveInRedis(item.getSn(),item.getTerminalTime()*1000);
                    if(!flag){
                        TerminalHistoryData historyData = new TerminalHistoryData();
                        historyData.setReadNumber(item.getReadNumber() == null ? null : new BigDecimal(item.getReadNumber()));
                        historyData.setReverseNumber(item.getReverseNumber() == null ? null : new BigDecimal(item.getReverseNumber()));
                        historyData.setInstantNumber(item.getInstantNumber() == null ? null : new BigDecimal(item.getInstantNumber()));
                        historyData.setTotalFlow(item.getTotalFlow() == null ? null : new BigDecimal(item.getTotalFlow()));
    //                    historyData.setPressure(item.getPressure() == null ? null : new BigDecimal(item.getPressure()));//没压力
                        historyData.setLocationNb(terminalsDb.getLocationNb());
                        historyData.setTerminalId(terminalsDb.getTerminalId());
                        historyData.setTenantId(terminalsDb.getTenantId());
    //                    historyData.setTerminalCsq(item.getCsq());
    //                    historyData.setValveState(item.getValveState());
                        historyData.setReadDate(new Date(item.getTerminalTime()*1000));
                        list.add(historyData);
                        redisCache.SortedSetAdd(RedisKeyName.backupDB_HistoricalData+factoryCode+"_" + item.getSn(),JSONObject.from(historyData), item.getTerminalTime());

                    }
                }

                if(!list.isEmpty()){
                    //历史数据插入到数据
                    terminalHistoryDataService.batchinsertTerminalHistoryData(list);
                    list.sort(Comparator.comparing(TerminalHistoryData::getReadDate).reversed());//根据readDate降序
                    if(null!=valveStatus){
                        list.get(0).setValveState((long) valveStatus);//最新的一条历史数据，设置阀门状态
                    }
                    realtimegatherdataService.synchronousMostValueFromRealtimegatherdata(list.get(0), terminalsDb);
                }

                }
        }catch (Exception e){
            log.error("解析失败：{}", ExceptionUtil.getExceptionMessage(e));
        }

    }


    //校验该条历史数据是否重复
    public Boolean  checkIsSaveInRedis(String communicateAddress, Long time)
    {
        Boolean flag = false;
//        long time = ParseDataUntils.ConvertDateTimeToInt(startTime);
        Set<TerminalHistoryData> list = redisCache.SortedSetRangeByRank(RedisKeyName.backupDB_HistoricalData+factoryCode+"_" + communicateAddress, time, time);
        if(null != list && list.size() > 0)
        {
            flag = true;
        }
        return flag;
    }
}
